package cn.mengfly.testnrz.crazyIm.nettyserver;

import cn.mengfly.testnrz.im.codec.ProtobufEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.Data;

import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * IM 转发器
 *
 * @author Mengfly
 */
@Data
public class WorkerReSender {

    private Channel channel;
    private ImNode remoteNode;

    private boolean connectFlag = false;

    private Bootstrap b;
    private EventLoopGroup g;


    private GenericFutureListener<ChannelFuture> connectListener = new GenericFutureListener<ChannelFuture>() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                connectFlag = true;
                System.out.println("分布式节点 " + remoteNode + "连接成功");
                channel = future.channel();
            } else {
                connectFlag = false;
                System.out.println("连接失败，在10秒后准备重试连接");
                g.schedule(() -> doConnect(), 10, TimeUnit.SECONDS);

            }
        }
    };

    public WorkerReSender(ImNode n) {
        this.remoteNode = n;
        b = new Bootstrap();
        g = new NioEventLoopGroup();
    }

    public void doConnect() {
        String host = remoteNode.getHost();
        int port = Integer.parseInt(remoteNode.getPort());
        try {

            if (b != null && b.config().group() == null) {
                b.group(g);
                b.channel(NioSocketChannel.class);
                b.option(ChannelOption.SO_KEEPALIVE, true);
                b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
                b.remoteAddress(host, port);
                b.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new ProtobufEncoder());
                    }
                });
                System.out.println(new Date() + " 开始连接分布式节点" + remoteNode.toString());
                final ChannelFuture f = b.connect();
                f.addListener(connectListener);
            } else if (b != null && b.config().group() != null) {
                System.out.println(new Date() + " 再一次连接分布式节点" + remoteNode.toString());
                final ChannelFuture connect = b.connect();
                connect.addListener(connectListener);

            }
        } catch (Exception e) {
            System.out.println("客户端连接失败！" + e.getMessage());
        }
    }

    public void writeAndFlush(Object pkg) {
        if (!connectFlag) {
            System.out.println("分布式节点未连接:" + remoteNode.toString());
            return;
        }
        channel.writeAndFlush(pkg);
    }

    public void disConnect() {
        final ChannelFuture close = channel.close();
        close.addListener(future -> {
            if (future.isSuccess()) {
                System.out.println("已断开分布式节点连接：" + remoteNode);
            }
        });
    }
}
